package com.example.hbase_server.service;

import com.example.hbase_server.Utils;
import com.example.hbase_server.pojo.MovieRating;
import com.example.hbase_server.pojo.Ratings;
import com.example.hbase_server.pojo.TempRating;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.*;
import org.apache.flink.table.api.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Service;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.commons.csv.*;
import java.io.FileReader;
import java.io.IOException;
import java.io.Reader;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;

import java.io.IOException;
import java.lang.reflect.Array;
import java.math.BigInteger;
import java.util.*;

import static com.example.hbase_server.C.*;

@Service
public class Func {

    public static Connection connection;
    public static Admin admin;
    public static StreamExecutionEnvironment streamEnv;
    public static StreamTableEnvironment tableEnv;
    public static Map<String, Object> popularMovie;
    public static void init() throws Exception {
        Configuration config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "bigdata");
        config.set("hbase.zookeeper.property.clientPort", "2181");
        connection = ConnectionFactory.createConnection(config);
        admin = connection.getAdmin();
        streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        tableEnv = StreamTableEnvironment.create(streamEnv);
        popularMovie = new HashMap<String, Object>();
        Utils.createAllTable(tableEnv);

        System.out.println("正在初始化");
        if(!Utils.judgmentTable(admin,"movies")){
            System.out.println("movies开始导入");
            Utils.createTable(admin,"movies",new String[]{"movieInfo"});
            Utils.createCSVTable(tableEnv,MOVIE_Path, """
                    create table moviesCSV (
                        movieId string,
                        title string,
                        genres string
                    )
            """);
            tableEnv.executeSql("INSERT INTO moviesTable (movieId, movieInfo) SELECT movieId, ROW (title, genres) FROM moviesCSV WHERE title <> 'title'").await();
            System.out.println("movies导入完成");
        }
        if(!Utils.judgmentTable(admin,"ratings")){
            System.out.println("ratings开始导入");
            Utils.createTable(admin,"ratings",new String[]{"ratingInfo"});
            Utils.createCSVTable(tableEnv,Rating_Path, """
                    create table ratingsCSV (
                        userId string,
                        movieId string,
                        rating DOUBLE,
                        ttl BIGINT
                    )
            """);

            tableEnv.executeSql("INSERT INTO ratingsTable (userId_movieId, ratingInfo) SELECT CONCAT(userId, '_', movieId), ROW (rating, ttl) FROM ratingsCSV").await();

            System.out.println("ratings导入完成");
        }
        if(!Utils.judgmentTable(admin,"movieRating")){
            System.out.println("movieRating开始导入");
            Utils.createTable(admin,"movieRating",new String[]{"ratingInfo"});
            Utils.createCSVTable(tableEnv,Rating_Path, """
                    create table ratingsCSV (
                        userId string,
                        movieId string,
                        rating DOUBLE,
                        ttl BIGINT
                    )
            """);

            tableEnv.executeSql("INSERT INTO movieRatingTable (movieId, ratingInfo) SELECT movieId, ROW (Count(rating), SUM(rating), AVG(rating)) FROM ratingsCSV GROUP BY movieId").await();

            System.out.println("movieRating导入完成");
        }
        if(!Utils.judgmentTable(admin,"tempRating")){
            Utils.createTable(admin,"tempRating",new String[]{"ratingInfo"});
        }
        System.out.println("已连接");
        Timer timer = new Timer();
        TimerTask task = new TimerTask() {
            @Override
            public void run() {
                try {
                    StreamTableEnvironment _tnv = StreamTableEnvironment.create(streamEnv);
                    long localTime = System.currentTimeMillis() / 1000 -600;
                    Table t = tableEnv.sqlQuery(String.format("SELECT movieId, rating, ttl AS `timestamp` FROM tempRatingTable where ttl >= %s",localTime));
                    List<Tuple4<String, Double, Long, Integer>> source = tableEnv.toDataStream(t, TempRating.class).map(new MapFunction<TempRating, Tuple4<String, Double, Long, Integer>>() {
                        @Override
                        public Tuple4<String, Double, Long, Integer> map(TempRating ratings) throws Exception {
                            return new Tuple4<String, Double, Long, Integer>(ratings.getMovieId(), ratings.getRating(), ratings.getTimestamp(), 1);
                        }
                    }).keyBy(new KeySelector<Tuple4<String, Double, Long, Integer>, Object>() {
                        @Override
                        public Object getKey(Tuple4<String, Double, Long, Integer> stringDoubleLongIntegerTuple4) throws Exception {
                            return stringDoubleLongIntegerTuple4.f0;
                        }
                    }).sum(3).executeAndCollect(99999);
                    source.sort(new Comparator<Tuple4<String, Double, Long, Integer>>() {
                        @Override
                        public int compare(Tuple4<String, Double, Long, Integer> o1, Tuple4<String, Double, Long, Integer> o2) {
                            return o2.f3 - o1.f3;
                        }
                    });
                    ArrayList<String> ten = new ArrayList<String>();
                    Map<String, Object> tempMovie = new HashMap<String, Object>();
                    source.forEach((Tuple4<String, Double, Long, Integer> item) ->{
                        if(ten.size()<10 && !ten.contains(item.f0)){
                            try {
                                ten.add(item.f0);
                                Table tTitle = tableEnv.sqlQuery(String.format("SELECT title FROM moviesTable where movieId = '%s'",item.f0));
                                List<String> title = tableEnv.toDataStream(tTitle,String.class).executeAndCollect(999);
                                Map<String, Object> oneMovie = new HashMap<String, Object>();
                                oneMovie.put("title",title.get(0));
                                oneMovie.put("rating",item.f1);
                                tempMovie.put(title.get(0),oneMovie);
                            } catch (Exception e) {
                                throw new RuntimeException(e);
                            }
                        }
                    });
                    popularMovie=tempMovie;
                }catch (Exception e){
                    e.printStackTrace();
                }

            }

        };
        timer.schedule(task,0,20000);
    }


    public static void close(){
        System.out.println("正在关闭连接");
        try{
            if(admin != null){
                admin.close();
            }
            if(null != connection){
                connection.close();
            }
            if (streamEnv != null){
                streamEnv.close();
            }
        }catch (IOException e){
            e.printStackTrace();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public static Map<String, Object> addComment(String userId, String movieId, Double rating) throws Exception {
        Table t= tableEnv.sqlQuery(String.format("SELECT movieId, ratingCount,ratingSum,rating FROM movieRatingTable where movieId = '%s' LIMIT 1",movieId));
        List<MovieRating> tList =tableEnv.toDataStream(t, MovieRating.class).executeAndCollect(99999999);
        DataStream<Ratings> ratingSource = streamEnv.fromElements(new Ratings(userId+"_"+movieId,rating,System.currentTimeMillis()/1000));
        DataStream<MovieRating> movieRatingSource = streamEnv.fromElements(
                        new MovieRating(tList.get(0).getMovieId(),
                                tList.get(0).getRatingCount(),
                                tList.get(0).getRatingSum(),
                                tList.get(0).getRating()));
        DataStream<MovieRating> mapSource = movieRatingSource.map(new MapFunction<MovieRating,MovieRating>() {
            @Override
            public MovieRating map(MovieRating row) throws Exception {
                row.ratingSum+=rating;
                row.rating=row.ratingSum / ++row.ratingCount;
                return row;
            }
        });
        ratingSource.addSink(new SinkFunction<Ratings>(){
            @Override
            public void invoke(Ratings value) throws Exception {
                SinkFunction.super.invoke(value);
                org.apache.hadoop.hbase.client.Table tb = connection.getTable(TableName.valueOf("ratings"));
                Put put = new Put(Bytes.toBytes(value.getUserId_movieId()));
                put.addColumn(Bytes.toBytes("ratingInfo"), Bytes.toBytes("rating"), Bytes.toBytes(value.getRating()));
                put.addColumn(Bytes.toBytes("ratingInfo"), Bytes.toBytes("ttl"), Bytes.toBytes(value.getTimestamp()));
                tb.put(put);
                tb.close();
            }
        });
        ratingSource.addSink(new SinkFunction<Ratings>(){
            @Override
            public void invoke(Ratings value) throws Exception {
                SinkFunction.super.invoke(value);
                org.apache.hadoop.hbase.client.Table tb = connection.getTable(TableName.valueOf("tempRating"));
                String movieId = value.getUserId_movieId().split("_")[1];
                Put put = new Put(Bytes.toBytes(value.getTimestamp()));
                put.addColumn(Bytes.toBytes("ratingInfo"), Bytes.toBytes("rating"), Bytes.toBytes(value.getRating()));
                put.addColumn(Bytes.toBytes("ratingInfo"), Bytes.toBytes("movieId"), Bytes.toBytes(movieId));
                tb.put(put);
                tb.close();
            }
        });
        mapSource.addSink(new SinkFunction<MovieRating>(){
            @Override
            public void invoke(MovieRating value) throws Exception {
                SinkFunction.super.invoke(value);
                org.apache.hadoop.hbase.client.Table tb = connection.getTable(TableName.valueOf("movieRating"));
                Put put = new Put(Bytes.toBytes(value.getMovieId()));
                put.addColumn(Bytes.toBytes("ratingInfo"), Bytes.toBytes("ratingCount"), Bytes.toBytes(value.getRatingCount()));
                put.addColumn(Bytes.toBytes("ratingInfo"), Bytes.toBytes("ratingSum"), Bytes.toBytes(value.getRatingSum()));
                put.addColumn(Bytes.toBytes("ratingInfo"), Bytes.toBytes("rating"), Bytes.toBytes(value.getRating()));
                tb.put(put);
                tb.close();
            }
        });
        streamEnv.execute();
        Map<String, Object> map = new HashMap<String, Object>();
        map.put("status","ok");
        return map;
    }

    public static Map<String, Object> getRating(String movieId) throws Exception{
        Table t= tableEnv.sqlQuery(String.format("SELECT movieId, ratingCount,ratingSum,rating FROM movieRatingTable where movieId = '%s' LIMIT 1",movieId));
        List<MovieRating> tList =tableEnv.toDataStream(t, MovieRating.class).executeAndCollect(99999999);
        Map<String, Object> map = new HashMap<String, Object>();
        map.put("movieId",tList.get(0).getMovieId());
        map.put("ratingCount",tList.get(0).getRatingCount());
        map.put("ratingSum",tList.get(0).getRatingSum());
        map.put("rating",tList.get(0).getRating());
        return map;
    }

    public static Map<String, Object> getTenMovie() throws Exception{
        return popularMovie;
    }

    public static Map<String, Object> getMovies(String startRow, String pageSize) throws IOException{
        int pageSizeNum =  Integer.parseInt(pageSize);
        if(pageSize.equals("")){
            pageSizeNum = 3;
        }
        if(startRow.equals("")){
            startRow ="1";
        }
        org.apache.hadoop.hbase.client.Table table = connection.getTable(TableName.valueOf("movies"));
        Filter filter = new PageFilter(pageSizeNum + 1);
        Scan scan = new Scan();
        scan.setFilter(filter);
        scan.setStartRow(startRow.getBytes());
        ResultScanner result = table.getScanner(scan);
        String newStartRow = "1";
        Map<String, Object> map = new HashMap<String, Object>();
        for(Result r:result){
            Map<String, Object> mapInfo = new HashMap<String, Object>();
            String movieId = new String(r.getRow());
            String title = Bytes.toString(r.getValue(Bytes.toBytes("movieInfo"),Bytes.toBytes("title")));
            String genres = Bytes.toString(r.getValue(Bytes.toBytes("movieInfo"),Bytes.toBytes("genres")));

            mapInfo.put("title", title);
            mapInfo.put("genres", genres);

            map.put(movieId, mapInfo);
            newStartRow = new String(r.getRow());
        }
        if(map.size() > pageSizeNum){
            map.remove(newStartRow);
        }
        map.put("startRow", newStartRow);
        table.close();
        return map;
    }

    public static Map<String, Object> searchMovieRating(String movieName) throws Exception{
        Table t= tableEnv.sqlQuery(String.format("SELECT a.rating FROM movieRatingTable a INNER JOIN  moviesTable b ON a.movieId = b.movieId where b.title = '%s' LIMIT 1",movieName));
        List<Double> tList =tableEnv.toDataStream(t, Double.class).executeAndCollect(999);
        Map<String, Object> map = new HashMap<String, Object>();
        map.put("rating",tList.get(0));
        return map;
    }
}
